package chen.bupt.mapreduce.usermodel;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import chen.bupt.constant.Constants;
import chen.bupt.util.HDFSFileUtils;

public class UserVSMJob {
	private static final String featurePath = Constants.FEATURE_TERM_PATH;
	private static final String inputPath = Constants.USER_FEATURE_PATH;
	private static final String outputPath = Constants.USER_VECTOR;
	private static Map<String, Integer> map = new HashMap<String, Integer>();

	// feature load
	static {
		try {
			int i = 0;
			Configuration conf = new Configuration();
			conf.set("fs.default.name", "hdfs://10.1.1.26:54310");
			FileSystem fs = FileSystem.get(conf);
			FSDataInputStream fis = fs.open(new Path(featurePath));
			BufferedReader br = new BufferedReader(new InputStreamReader(fis));
			String list = br.readLine();
			String[] words = list.split(";");
			for (String word : words) {
				if (!"".equals(word)) {
					map.put(word, i);
					i++;
				}
			}
			br.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static class Map1 extends Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			if (!HDFSFileUtils.isInputFile(context))
				return;
			int index = value.toString().indexOf("|");
			if(index==-1)return;
			context.write(new Text(value.toString().substring(0, index)),
					new Text(value.toString().substring(index + 1)));
		}
	}

	private static class WordScore {
		public int index;
		public String score;

		public WordScore(int i, String s) {
			index = i;
			score = s;
		}

		public String toString() {
			return index + ";" + score;
		}
	}

	private static class MyCompatator implements Comparator<WordScore> {

		@Override
		public int compare(WordScore o1, WordScore o2) {

			return o1.index - o2.index;
		}

	}

	public static class Reduce1 extends Reducer<Text, Text, Text, Text> {
		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			List<WordScore> vector = new ArrayList<WordScore>();
			for (Text text : values) {
				String[] words = text.toString().split("\\|");
				for (String wordScore : words) {
					String word = wordScore.split(";")[0];
					String score = wordScore.split(";")[1];
					Integer ind = map.get(word);
					if (ind == null)
						continue;
					WordScore ws = new WordScore(ind, score);
					vector.add(ws);
				}
			}
			Collections.sort(vector, new MyCompatator());
			StringBuffer sb = new StringBuffer();
			for (WordScore ws : vector) {
				sb.append(ws.toString()+"|");
			}
			context.write(new Text(key), new Text(sb.toString()));
		}
	}

	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		conf.set("mapred.task.timeout", "0");
		Job job = new Job(conf,"VSMJob");
		job.setJarByClass(UserVSMJob.class);
		job.setMapperClass(Map1.class);
		job.setReducerClass(Reduce1.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		Path input = new Path(inputPath);
		Path output = new Path(outputPath);
		HDFSFileUtils.deleteFile(output, conf);
		FileInputFormat.addInputPath(job, input);
		FileOutputFormat.setOutputPath(job, output);
		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
